package sg.edu.nus.iss.stockex;

import java.io.IOException;
import java.text.DateFormat;
import java.text.ParseException;
import java.util.*;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class StockService {
	public static class Reduce extends
	Reducer<Text, Stock, Text, Text> {
		
		public void reduce(Text key, Iterable<Stock> values,
			Context context) throws IOException, InterruptedException {
		int sum = 0;
		double maxValue = 0;
		double minValue = Double.MAX_VALUE;
		long volume = 0;
		Text stockText = new Text();
	//	stockText.set("asdf");
		
		for(Stock stock : values)
		{
			if(stock.getHighPrice()>maxValue)
			{
				maxValue = stock.getHighPrice();
			}
			
			if(stock.getLowPrice() < minValue)
			{
				minValue = stock.getLowPrice();
			}
			volume +=stock.getVolume();
		}
	
		String data = "max: " + maxValue + " min: " + minValue + " volume: " + volume;
		stockText.set(data);
		context.write(key, stockText);
		}
	}
	
	public static class Map extends Mapper<LongWritable, Text, Text, Stock> {
		
//	private final static Company one = new Company();
	private Text word = new Text();

	public void map(LongWritable key, Text value, Context context)
			throws IOException, InterruptedException {
		try{
			String line = value.toString();
			System.out.println("Line is: "+line);
			Stock stock = Stock.parseString(line);
			word.set(stock.getCode());
			context.write(word, stock);
		}
		catch(Exception ex)
		{
			ex.printStackTrace();
		}
	}
}
	public static void startJob(int quarter, int year) throws Exception
	{
		Configuration conf = new Configuration();
		
		conf.setInt(CONFIG.QUARTER, quarter);
		conf.setInt(CONFIG.YEAR,year);
		Job job = new Job(conf, "StockEx");
		// output of the Map class
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(Stock.class);
		// output of the job..
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);

		job.setMapperClass(Map.class);
		job.setReducerClass(Reduce.class);

		job.setInputFormatClass(TextInputFormat.class);
		job.setOutputFormatClass(TextOutputFormat.class);

		FileInputFormat.addInputPath(job, new Path("input"));
		// Erase previous run output (if any)
		FileSystem.get(conf).delete(new Path("output"), true);
		FileOutputFormat.setOutputPath(job, new Path("output"));

		job.waitForCompletion(true);
	
	}
	public static void main(String[] args) throws Exception {
		
		int quarter = 2;
		int year = 2012;
		startJob( quarter, year);

	}
}
